package com.g2room.queue.kafka;

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
public class KafkaProducer {

	public static String topic = "gxg001_p3-0";
	//第一个类型代表key的类型，第二个代表消息的类型
	public static Producer<String, String> producer;
	
	public KafkaProducer() {
		if(producer == null) producer = new Producer<String, String>(new ProducerConfig(getProperties()));
	}
	
	/**
	 * 发送消息，如果发送成功，则返回true，反之返回false
	 * @param msg	待发送的消息
	 * @return
	 */
	public boolean sendMsg(String mid, String msg) {
		boolean rs = false;
		if(msg == null) return rs;
		try {
			
			// 不指定key分配方式的调用
//            producer.send(new KeyedMessage<String, String>(topic, str));
			// KeyedMessage第一个参数代表key的类型，第二个参数代表message的类型
			producer.send(new KeyedMessage<String, String>(topic, mid, msg));
			rs = true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return rs;
	}
	
	public Properties getProperties() {
		Properties properties = new Properties();
        // 配置ZK连接
		properties.put("zookeeper.connect", "10.3.246.63:6381,10.3.246.60:6381");
		 //zk连接超时
		properties.put("zookeeper.session.timeout.ms", "4000");
		properties.put("zookeeper.sync.time.ms", "200");
		// 配置序列化类
		properties.put("serializer.class", StringEncoder.class.getName());
//        //配置value的序列化类
//        properties.put("serializer.class", "kafka.serializer.StringEncoder");
        //此处配置的是kafka的集群
		properties.put("metadata.broker.list", "10.3.246.63:9092,10.3.246.60:9092");
//		0，意味着producer永远不会等待一个来自broker的ack，这就是0.7版本的行为。这个选项提供了最低的延迟，但是持久化的保证是最弱的，当server挂掉的时候会丢失一些数据。
//		1，意味着在leader replica已经接收到数据后，producer会得到一个ack。这个选项提供了更好的持久性，因为在server确认请求成功处理后，client才会返回。如果刚写到leader上，还没来得及复制leader就挂了，那么消息才可能会丢失。
//		-1，意味着在所有的ISR都接收到数据后，producer才得到一个ack。这个选项提供了最好的持久性，只要还有一个replica存活，那么数据就不会丢失。
		properties.put("request.required.acks","1");
		return properties;
	}

	public static void main(String[] args) {
		KafkaProducer kp = new KafkaProducer();
		String 	msg = "1234";
		for(int i =0; i<100000; i++)
		System.out.println(kp.sendMsg(i+"", msg + i));
		producer.close();
	}

}
